How come we need to change this from a list to set? I looked at how get_prefixes_from_globs used in S3 and it looks like we just check it in an if statement followed by a for each loop.

We loop over the prefixes and submit a request to S3 for each one. I wanted to avoid submitting another request to S3 in the event that there were duplicates.

I noticed file_read_mode is only used with the parser implementation file itself. So it is not strictly necessary that it be exposed in the interface. Is the idea in making it part of the interface to be an indicator that future file parser types need to provide an implementation to this?

Yes, exactly.

How is this being called? I didn't see it usage in the PR or when I searched in IntelliJ. The only concern here is that these statically defined constants don't work that well if we decide to move in the direction of custom file formats per source

Oops, that was a remnant of something that I ultimately deleted. Good catch, removed.

nit for consistency let's have everything have a title so we dont rely on pydantics auto capitalizing

let's add a comment here with our prior conversation to why we have to check this value for type safety. At face value without context it might seem redundant.

@clnoll I feel like this is worth discussing with other reviewers in the PR because it's probably one of the more confusing aspects. Since we're trying to force a specific subclass of AbstractFileBasedSpec (in this case S3 config) which makes sense. But it causes some type check warnings since the AbstractFileBasedStreamReader interface is generalized. I'm not sure of a better solution either

Sure, @brianjlai, added a comment.

To give some context to other reviewers - the basic background here is that config information is needed by both the FileBasedSource and the StreamReader, but for different reasons; FileBasedSource needs it to configure streams, and the StreamReader needs it so that it knows how to, e.g. authenticate with the storage system.

In this PR, I'm allowing FileBasedSource to read the config from disk and parse it (like normal), and once parsed, the source sets the config on the StreamReader. However, the type of config accepted by FileBasedSource is AbstractFileBasedSpec, and it only cares about keys that are source-agnostic, whereas StreamReader cares about keys that are specific to the 3rd party that it's reading from. For example, S3 will be looking for a aws_access_token_id key, but FileBasedSource won't itself need that.

So that leads us to this situation, where the S3 StreamReader's config setter requires a config of type (S3) Config, but the interface config setter takes an AbstractFileBasedSpec, which doesn't have S3-specific keys. To solve this, we assert that we were given the correct Config type for our type of StreamReader.

One alternative route that I went down involved reading in the config prior to the initialization of the Source, so that it could be given to the StreamReader as an argument. Since the Source requires a StreamReader, it could get the config off of that. This is somewhat undesirable because we end up reading the config twice (because AbstractSource still reads it deep within the CDK code), and it also deviates from the pattern of letting the Source validate the config, which may have error handling behavior that we want to keep. For those two reasons I'd prefer to keep the code as-is. But I'm open to other opinions.

Curious, does mypy checks not flag this since the interface method expects AbstractFileBasedSpec even though Config is a subclass of it

Yeah mypy doesn't seem to mind about it.

are we expecting other kwargs? I'm not sure I see the benefit of putting Bucket behind kwargs and passing it to _page. It just makes it more opaque what it contains in that method.

Also related nit, if we do use kwargs, I'm used to seeing it as the last parameter in the _page function

We add a ContinuationToken in _page, so I moved the creation of the kwargs dict there.

Although len(seen) would be correct for both flows, wouldn't total_n_keys always be 1 if there are no prefixes. And for prefixes, it would always log 0 since it never changes total_n_keys. We count total_n_keys_for_prefix within _page() but we only print the extra log in that method.

What count are we trying to display here and would this extra log be duplicative with the ones in _page()?

Oops yep total_n_keys should have been in each of the for remote_file in ... loops. Fixed.

Would it make sense to stand up the method stubs for the S3FileSource instead of using FileBasedSource in this PR to help if the next set of PRs easier so there aren't conflicts modifying that central class?

Could do, but I'm not anticipating that we'll need new methods on FileBasedSource so wasn't planning to create a separate S3FileSource (most of the customization occurs in the objects that are passed into FileBasedSource). Does that make sense to you or does it feel like there should be a dedicated S3FileSource?

So I was sketching out the work for the S3 legacy config transformer to the new config. A lot of the config validation against the spec is done automatically in the entrypoint. So the potentially easiest way to transform legacy manifests before we perform the validation is to have an S3FileSource override BaseConnector.read_config().

I'm still toying with ideas about how to do it best, but so far that was the simplest way to avoid having to manage multiple specs and versions of the config. But if you don't think your work will require it, then we won't conflict anyway

Yeah what I'm doing now won't require it, so I think I'll let you make that change.

This will need to be released before the airbyte-integrations as it will not be available for the source. Consequence: if someone needs to release source-s3, the tests will fail

This was actually moved, in a separate PR, so I've deleted it. This PR now just consists of the S3 connector changes.

I'm probably missing something: why do we set client_kv_args = {"config": client_config} if we update client_kv_args["config"] right after with ClientConfig(s3={"addressing_style": "auto"})?

You're right, this is awkward (a remnant of copy/pasting & an incomplete refactor). Cleaned it up.